--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 2ce29a6488e2af4351d59ad7246dc5a9927cee51
Parents : 0ae45b4
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-01-14T13:02:33-06:00
ruff format and fixes
Changes
13 files changed, 201 insertions(+), 82 deletions(-)
Diff
diff --git a/tests/backend/test_auto_propagation.py b/tests/backend/test_auto_propagation.py
index 6eb6cc99..bc1dcc9c 100644
--- a/tests/backend/test_auto_propagation.py
+++ b/tests/backend/test_auto_propagation.py
@@ -1,6 +1,8 @@
from unittest.mock import MagicMock, patch
+
import pytest
import RNS
+
from meshchatx.src.backend.auto_propagation_manager import AutoPropagationManager
@@ -54,7 +56,7 @@ async def test_auto_propagation_logic():
# Should have selected aaaa1111
app.set_active_propagation_node.assert_called_with("aaaa1111", context=context)
config.lxmf_preferred_propagation_node_destination_hash.set.assert_called_with(
- "aaaa1111"
+ "aaaa1111",
)
# 3. Test switching to better node
diff --git a/tests/backend/test_auto_propagation_api.py b/tests/backend/test_auto_propagation_api.py
index f89d7bee..04adbace 100644
--- a/tests/backend/test_auto_propagation_api.py
+++ b/tests/backend/test_auto_propagation_api.py
@@ -3,8 +3,10 @@ import json
import shutil
import tempfile
from unittest.mock import MagicMock, patch
+
import pytest
import RNS
+
from meshchatx.meshchat import ReticulumMeshChat
@@ -78,7 +80,7 @@ async def test_auto_propagation_api(mock_rns_minimal, temp_dir):
mock_request = MagicMock()
mock_request.json = MagicMock(return_value=asyncio.Future())
mock_request.json.return_value.set_result(
- {"lxmf_preferred_propagation_node_auto_select": True}
+ {"lxmf_preferred_propagation_node_auto_select": True},
)
response = await patch_handler(mock_request)
@@ -90,7 +92,7 @@ async def test_auto_propagation_api(mock_rns_minimal, temp_dir):
mock_request = MagicMock()
mock_request.json = MagicMock(return_value=asyncio.Future())
mock_request.json.return_value.set_result(
- {"lxmf_preferred_propagation_node_auto_select": False}
+ {"lxmf_preferred_propagation_node_auto_select": False},
)
response = await patch_handler(mock_request)
diff --git a/tests/backend/test_csp_logic.py b/tests/backend/test_csp_logic.py
index da61cc46..b623056e 100644
--- a/tests/backend/test_csp_logic.py
+++ b/tests/backend/test_csp_logic.py
@@ -1,4 +1,4 @@
-from unittest.mock import MagicMock, patch, AsyncMock
+from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import RNS
@@ -42,7 +42,7 @@ async def test_csp_header_logic(mock_rns_minimal, tmp_path):
# Mock the config values
app_instance.config.csp_extra_connect_src.set("https://api.example.com")
app_instance.config.map_tile_server_url.set(
- "https://tiles.example.com/{z}/{x}/{y}.png"
+ "https://tiles.example.com/{z}/{x}/{y}.png",
)
# Mock a request and handler
@@ -99,12 +99,16 @@ async def test_config_update_csp(mock_rns_minimal, tmp_path):
# To avoid the JSON serialization error of MagicMock in get_config_dict,
# we mock get_config_dict to return a serializable dict.
- with patch.object(
- app_instance, "get_config_dict", return_value={"status": "ok"}
+ with (
+ patch.object(
+ app_instance,
+ "get_config_dict",
+ return_value={"status": "ok"},
+ ),
+ patch.object(app_instance, "send_config_to_websocket_clients"),
):
- with patch.object(app_instance, "send_config_to_websocket_clients"):
- response = await config_update_handler(request)
- assert response.status == 200
+ response = await config_update_handler(request)
+ assert response.status == 200
assert (
app_instance.config.csp_extra_connect_src.get()
diff --git a/tests/backend/test_display_name_and_telemetry.py b/tests/backend/test_display_name_and_telemetry.py
index 5b14cc30..6a637fe3 100644
--- a/tests/backend/test_display_name_and_telemetry.py
+++ b/tests/backend/test_display_name_and_telemetry.py
@@ -1,14 +1,15 @@
import base64
from unittest.mock import MagicMock, patch
+
import LXMF
+import RNS.vendor.umsgpack as msgpack
+
from meshchatx.src.backend.meshchat_utils import parse_lxmf_display_name
from meshchatx.src.backend.telemetry_utils import Telemeter
-import RNS.vendor.umsgpack as msgpack
def test_parse_lxmf_display_name_bug_fix():
- """
- Test that parse_lxmf_display_name handles both bytes and strings
+ """Test that parse_lxmf_display_name handles both bytes and strings
in the msgpack list, fixing the 'str' object has no attribute 'decode' bug.
"""
# 1. Test with bytes (normal case)
@@ -34,9 +35,7 @@ def test_parse_lxmf_display_name_bug_fix():
def test_lxmf_telemetry_decoding():
- """
- Test decoding of LXMF telemetry fields.
- """
+ """Test decoding of LXMF telemetry fields."""
# Create some dummy telemetry data
ts = 1736264575
lat, lon = 52.5200, 13.4050
@@ -68,9 +67,7 @@ def test_lxmf_telemetry_decoding():
def test_lxmf_telemetry_mapping_in_app():
- """
- Test how the app handles telemetry fields from an LXMF message.
- """
+ """Test how the app handles telemetry fields from an LXMF message."""
# Mock lxmf_message
lxmf_message = MagicMock(spec=LXMF.LXMessage)
source_hash = b"\x01" * 32
@@ -79,7 +76,8 @@ def test_lxmf_telemetry_mapping_in_app():
ts = 1736264575
packed_telemetry = Telemeter.pack(
- time_utc=ts, location={"latitude": 1.23, "longitude": 4.56}
+ time_utc=ts,
+ location={"latitude": 1.23, "longitude": 4.56},
)
lxmf_message.get_fields.return_value = {LXMF.FIELD_TELEMETRY: packed_telemetry}
diff --git a/tests/backend/test_fuzzing.py b/tests/backend/test_fuzzing.py
index 37686e78..97d7f460 100644
--- a/tests/backend/test_fuzzing.py
+++ b/tests/backend/test_fuzzing.py
@@ -106,8 +106,9 @@ def test_display_name_parsing_fuzzing(app_data_base64):
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
fields_data=st.dictionaries(
- st.integers(min_value=0, max_value=255), st.binary(min_size=0, max_size=1000)
- )
+ st.integers(min_value=0, max_value=255),
+ st.binary(min_size=0, max_size=1000),
+ ),
)
def test_lxmf_fields_parsing_fuzzing(fields_data):
"""Fuzz the parsing of LXMF message fields."""
diff --git a/tests/backend/test_fuzzing_telemetry.py b/tests/backend/test_fuzzing_telemetry.py
index d407ef19..3ca66279 100644
--- a/tests/backend/test_fuzzing_telemetry.py
+++ b/tests/backend/test_fuzzing_telemetry.py
@@ -1,4 +1,6 @@
-from hypothesis import given, settings, strategies as st
+from hypothesis import given, settings
+from hypothesis import strategies as st
+
from meshchatx.src.backend.telemetry_utils import Telemeter
# Strategies for telemetry data
@@ -75,7 +77,7 @@ def test_fuzz_from_packed_random_bytes(data):
values=st.one_of(st.integers(), st.text(), st.floats(), st.booleans()),
),
max_size=10,
- )
+ ),
)
def test_fuzz_command_parsing(commands):
# This simulates how commands are handled in meshchat.py
diff --git a/tests/backend/test_identity_restore.py b/tests/backend/test_identity_restore.py
index c0efe251..4398f68a 100644
--- a/tests/backend/test_identity_restore.py
+++ b/tests/backend/test_identity_restore.py
@@ -1,9 +1,10 @@
+import base64
import os
import shutil
import tempfile
-import base64
import unittest
from unittest.mock import MagicMock, patch
+
from meshchatx.src.backend.identity_manager import IdentityManager
@@ -19,7 +20,10 @@ class TestIdentityRestore(unittest.TestCase):
@patch("meshchatx.src.backend.identity_manager.DatabaseProvider")
@patch("meshchatx.src.backend.identity_manager.DatabaseSchema")
def test_restore_identity_from_bytes(
- self, mock_schema, mock_provider, mock_rns_identity
+ self,
+ mock_schema,
+ mock_provider,
+ mock_rns_identity,
):
# Setup mock identity
mock_id_instance = MagicMock()
@@ -48,7 +52,10 @@ class TestIdentityRestore(unittest.TestCase):
@patch("meshchatx.src.backend.identity_manager.DatabaseProvider")
@patch("meshchatx.src.backend.identity_manager.DatabaseSchema")
def test_restore_identity_from_base32(
- self, mock_schema, mock_provider, mock_rns_identity
+ self,
+ mock_schema,
+ mock_provider,
+ mock_rns_identity,
):
# Setup mock identity
mock_id_instance = MagicMock()
diff --git a/tests/backend/test_identity_switch.py b/tests/backend/test_identity_switch.py
index 32552f37..e4a25f80 100644
--- a/tests/backend/test_identity_switch.py
+++ b/tests/backend/test_identity_switch.py
@@ -76,7 +76,9 @@ def mock_rns():
)
stack.enter_context(
patch.object(
- MockIdentityClass, "from_bytes", return_value=mock_id_instance
+ MockIdentityClass,
+ "from_bytes",
+ return_value=mock_id_instance,
),
)
diff --git a/tests/backend/test_lxmf_icons.py b/tests/backend/test_lxmf_icons.py
index a7836912..b19b0242 100644
--- a/tests/backend/test_lxmf_icons.py
+++ b/tests/backend/test_lxmf_icons.py
@@ -92,7 +92,9 @@ def mock_rns():
)
stack.enter_context(
patch.object(
- MockIdentityClass, "from_bytes", return_value=mock_id_instance
+ MockIdentityClass,
+ "from_bytes",
+ return_value=mock_id_instance,
),
)
diff --git a/tests/backend/test_maintenance.py b/tests/backend/test_maintenance.py
index 9a9d3f93..3cbcc11a 100644
--- a/tests/backend/test_maintenance.py
+++ b/tests/backend/test_maintenance.py
@@ -1,7 +1,8 @@
import unittest
from unittest.mock import MagicMock
-from meshchatx.src.backend.database.messages import MessageDAO
+
from meshchatx.src.backend.database.announces import AnnounceDAO
+from meshchatx.src.backend.database.messages import MessageDAO
from meshchatx.src.backend.database.misc import MiscDAO
@@ -27,7 +28,8 @@ class TestMaintenance(unittest.TestCase):
# Test with aspect
self.announces_dao.delete_all_announces(aspect="test_aspect")
self.provider.execute.assert_called_with(
- "DELETE FROM announces WHERE aspect = ?", ("test_aspect",)
+ "DELETE FROM announces WHERE aspect = ?",
+ ("test_aspect",),
)
def test_delete_all_favourites(self):
@@ -38,7 +40,8 @@ class TestMaintenance(unittest.TestCase):
# Test with aspect
self.announces_dao.delete_all_favourites(aspect="test_aspect")
self.provider.execute.assert_called_with(
- "DELETE FROM favourite_destinations WHERE aspect = ?", ("test_aspect",)
+ "DELETE FROM favourite_destinations WHERE aspect = ?",
+ ("test_aspect",),
)
def test_delete_archived_pages(self):
diff --git a/tests/backend/test_property_based.py b/tests/backend/test_property_based.py
index 428ab781..0511c75f 100644
--- a/tests/backend/test_property_based.py
+++ b/tests/backend/test_property_based.py
@@ -1,21 +1,15 @@
-import math
-import struct
-import time
-from typing import Any
-import json
-import base64
import html
+import json
+import math
import pytest
-from hypothesis import given, strategies as st, HealthCheck, settings
+from hypothesis import HealthCheck, given, settings
+from hypothesis import strategies as st
+
from meshchatx.src.backend.colour_utils import ColourUtils
from meshchatx.src.backend.interface_config_parser import InterfaceConfigParser
from meshchatx.src.backend.lxmf_utils import convert_db_lxmf_message_to_dict
from meshchatx.src.backend.markdown_renderer import MarkdownRenderer
-from meshchatx.src.backend.nomadnet_utils import (
- convert_nomadnet_field_data_to_map,
- convert_nomadnet_string_data_to_map,
-)
from meshchatx.src.backend.meshchat_utils import (
parse_bool_query_param,
parse_lxmf_display_name,
@@ -23,18 +17,47 @@ from meshchatx.src.backend.meshchat_utils import (
parse_lxmf_stamp_cost,
parse_nomadnetwork_node_display_name,
)
-from meshchatx.src.backend.telemetry_utils import Telemeter, Sensor
-from RNS.vendor import umsgpack
+from meshchatx.src.backend.nomadnet_utils import (
+ convert_nomadnet_field_data_to_map,
+ convert_nomadnet_string_data_to_map,
+)
+from meshchatx.src.backend.telemetry_utils import Telemeter
# Strategies for telemetry data
-st_latitude = st.floats(min_value=-90, max_value=90, allow_nan=False, allow_infinity=False)
-st_longitude = st.floats(min_value=-180, max_value=180, allow_nan=False, allow_infinity=False)
-st_altitude = st.floats(min_value=-10000, max_value=100000, allow_nan=False, allow_infinity=False)
+st_latitude = st.floats(
+ min_value=-90,
+ max_value=90,
+ allow_nan=False,
+ allow_infinity=False,
+)
+st_longitude = st.floats(
+ min_value=-180,
+ max_value=180,
+ allow_nan=False,
+ allow_infinity=False,
+)
+st_altitude = st.floats(
+ min_value=-10000,
+ max_value=100000,
+ allow_nan=False,
+ allow_infinity=False,
+)
st_speed = st.floats(min_value=0, max_value=1000, allow_nan=False, allow_infinity=False)
-st_bearing = st.floats(min_value=-360, max_value=360, allow_nan=False, allow_infinity=False)
-st_accuracy = st.floats(min_value=0, max_value=655.35, allow_nan=False, allow_infinity=False)
+st_bearing = st.floats(
+ min_value=-360,
+ max_value=360,
+ allow_nan=False,
+ allow_infinity=False,
+)
+st_accuracy = st.floats(
+ min_value=0,
+ max_value=655.35,
+ allow_nan=False,
+ allow_infinity=False,
+)
st_timestamp = st.integers(min_value=0, max_value=2**32 - 1)
+
@given(
lat=st_latitude,
lon=st_longitude,
@@ -42,14 +65,14 @@ st_timestamp = st.integers(min_value=0, max_value=2**32 - 1)
speed=st_speed,
bear=st_bearing,
acc=st_accuracy,
- ts=st_timestamp
+ ts=st_timestamp,
)
def test_telemeter_location_roundtrip(lat, lon, alt, speed, bear, acc, ts):
packed = Telemeter.pack_location(lat, lon, alt, speed, bear, acc, ts)
assert packed is not None
unpacked = Telemeter.unpack_location(packed)
assert unpacked is not None
-
+
# Check with tolerance due to rounding/fixed point conversion in packing
assert math.isclose(unpacked["latitude"], lat, abs_tol=1e-6)
assert math.isclose(unpacked["longitude"], lon, abs_tol=1e-6)
@@ -60,6 +83,7 @@ def test_telemeter_location_roundtrip(lat, lon, alt, speed, bear, acc, ts):
assert math.isclose(unpacked["accuracy"], acc, abs_tol=1e-2)
assert unpacked["last_update"] == ts
+
@given(
time_utc=st_timestamp,
lat=st_latitude,
@@ -68,20 +92,29 @@ def test_telemeter_location_roundtrip(lat, lon, alt, speed, bear, acc, ts):
charging=st.booleans(),
rssi=st.integers(min_value=-150, max_value=0),
snr=st.integers(min_value=-20, max_value=20),
- q=st.integers(min_value=0, max_value=100)
+ q=st.integers(min_value=0, max_value=100),
)
-def test_telemeter_full_pack_roundtrip(time_utc, lat, lon, charge, charging, rssi, snr, q):
+def test_telemeter_full_pack_roundtrip(
+ time_utc,
+ lat,
+ lon,
+ charge,
+ charging,
+ rssi,
+ snr,
+ q,
+):
location = {"latitude": lat, "longitude": lon}
battery = {"charge_percent": charge, "charging": charging}
physical_link = {"rssi": rssi, "snr": snr, "q": q}
-
+
packed = Telemeter.pack(
time_utc=time_utc,
location=location,
battery=battery,
- physical_link=physical_link
+ physical_link=physical_link,
)
-
+
unpacked = Telemeter.from_packed(packed)
assert unpacked is not None
assert unpacked["time"]["utc"] == time_utc
@@ -93,17 +126,26 @@ def test_telemeter_full_pack_roundtrip(time_utc, lat, lon, charge, charging, rss
assert unpacked["physical_link"]["snr"] == snr
assert unpacked["physical_link"]["q"] == q
+
@given(hex_val=st.from_regex(r"^#?[0-9a-fA-F]{6}$"))
def test_colour_utils_hex_to_byte_array(hex_val):
result = ColourUtils.hex_colour_to_byte_array(hex_val)
assert len(result) == 3
-
+
# Verify manual conversion matches
clean_hex = hex_val.lstrip("#")
expected = bytes.fromhex(clean_hex)
assert result == expected
-@given(val=st.one_of(st.sampled_from(["1", "true", "yes", "on", "0", "false", "no", "off", "random"]), st.none()))
+
+@given(
+ val=st.one_of(
+ st.sampled_from(
+ ["1", "true", "yes", "on", "0", "false", "no", "off", "random"],
+ ),
+ st.none(),
+ ),
+)
def test_parse_bool_query_param(val):
result = parse_bool_query_param(val)
if val is None:
@@ -113,6 +155,7 @@ def test_parse_bool_query_param(val):
else:
assert result is False
+
@given(data=st.binary())
def test_parse_lxmf_display_name_robustness(data):
# This should never crash
@@ -121,6 +164,7 @@ def test_parse_lxmf_display_name_robustness(data):
except Exception as e:
pytest.fail(f"parse_lxmf_display_name crashed: {e}")
+
@given(data=st.binary())
def test_parse_lxmf_propagation_node_app_data_robustness(data):
# This should never crash
@@ -129,6 +173,7 @@ def test_parse_lxmf_propagation_node_app_data_robustness(data):
except Exception as e:
pytest.fail(f"parse_lxmf_propagation_node_app_data crashed: {e}")
+
@given(data=st.binary())
def test_parse_lxmf_stamp_cost_robustness(data):
# This should never crash
@@ -137,6 +182,7 @@ def test_parse_lxmf_stamp_cost_robustness(data):
except Exception as e:
pytest.fail(f"parse_lxmf_stamp_cost crashed: {e}")
+
@given(name=st.text())
def test_parse_nomadnetwork_node_display_name_robustness(name):
# This should never crash
@@ -145,6 +191,7 @@ def test_parse_nomadnetwork_node_display_name_robustness(name):
except Exception as e:
pytest.fail(f"parse_nomadnetwork_node_display_name crashed: {e}")
+
@given(packed=st.binary())
def test_telemeter_from_packed_robustness(packed):
# This should never crash
@@ -153,6 +200,7 @@ def test_telemeter_from_packed_robustness(packed):
except Exception as e:
pytest.fail(f"Telemeter.from_packed crashed: {e}")
+
@given(text=st.text())
def test_markdown_renderer_no_crash(text):
try:
@@ -160,6 +208,7 @@ def test_markdown_renderer_no_crash(text):
except Exception as e:
pytest.fail(f"MarkdownRenderer.render crashed: {e}")
+
@given(text=st.text())
def test_interface_config_parser_no_crash(text):
try:
@@ -167,40 +216,74 @@ def test_interface_config_parser_no_crash(text):
except Exception as e:
pytest.fail(f"InterfaceConfigParser.parse crashed: {e}")
+
# Strategy for a database message row
st_db_message = st.dictionaries(
- keys=st.sampled_from([
- "id", "hash", "source_hash", "destination_hash", "is_incoming",
- "state", "progress", "method", "delivery_attempts",
- "next_delivery_attempt_at", "title", "content", "fields",
- "timestamp", "rssi", "snr", "quality", "is_spam",
- "created_at", "updated_at"
- ]),
+ keys=st.sampled_from(
+ [
+ "id",
+ "hash",
+ "source_hash",
+ "destination_hash",
+ "is_incoming",
+ "state",
+ "progress",
+ "method",
+ "delivery_attempts",
+ "next_delivery_attempt_at",
+ "title",
+ "content",
+ "fields",
+ "timestamp",
+ "rssi",
+ "snr",
+ "quality",
+ "is_spam",
+ "created_at",
+ "updated_at",
+ ],
+ ),
values=st.one_of(
st.none(),
st.integers(),
st.floats(allow_nan=False, allow_infinity=False),
st.text(),
st.booleans(),
- st.binary().map(lambda b: b.hex())
- )
+ st.binary().map(lambda b: b.hex()),
+ ),
).filter(lambda d: "created_at" in d and "updated_at" in d)
+
@settings(suppress_health_check=[HealthCheck.too_slow])
@given(db_message=st_db_message)
def test_convert_db_lxmf_message_to_dict_robustness(db_message):
# Fill in missing required keys for the function
required_keys = [
- "id", "hash", "source_hash", "destination_hash", "is_incoming",
- "state", "progress", "method", "delivery_attempts",
- "next_delivery_attempt_at", "title", "content", "fields",
- "timestamp", "rssi", "snr", "quality", "is_spam",
- "created_at", "updated_at"
+ "id",
+ "hash",
+ "source_hash",
+ "destination_hash",
+ "is_incoming",
+ "state",
+ "progress",
+ "method",
+ "delivery_attempts",
+ "next_delivery_attempt_at",
+ "title",
+ "content",
+ "fields",
+ "timestamp",
+ "rssi",
+ "snr",
+ "quality",
+ "is_spam",
+ "created_at",
+ "updated_at",
]
for key in required_keys:
if key not in db_message:
db_message[key] = None
-
+
# Ensure fields is a valid JSON string if it's not None
if db_message["fields"] is not None:
try:
@@ -214,6 +297,7 @@ def test_convert_db_lxmf_message_to_dict_robustness(db_message):
# We expect some errors if data is really weird, but it shouldn't crash the whole thing
pass
+
@given(data=st.dictionaries(keys=st.text(), values=st.text()))
def test_convert_nomadnet_field_data_to_map(data):
result = convert_nomadnet_field_data_to_map(data)
@@ -221,8 +305,13 @@ def test_convert_nomadnet_field_data_to_map(data):
for k, v in data.items():
assert result[f"field_{k}"] == v
-@given(data=st.dictionaries(keys=st.text().filter(lambda x: "=" not in x and "|" not in x and x),
- values=st.text().filter(lambda x: "|" not in x)))
+
+@given(
+ data=st.dictionaries(
+ keys=st.text().filter(lambda x: "=" not in x and "|" not in x and x),
+ values=st.text().filter(lambda x: "|" not in x),
+ ),
+)
def test_convert_nomadnet_string_data_to_map_roundtrip(data):
# Construct string like key1=val1|key2=val2
input_str = "|".join([f"{k}={v}" for k, v in data.items()])
@@ -231,6 +320,7 @@ def test_convert_nomadnet_string_data_to_map_roundtrip(data):
for k, v in data.items():
assert result[f"var_{k}"] == v
+
@given(text=st.text())
def test_markdown_renderer_xss_protection(text):
# Basic check: if we use <script>, it should be escaped
@@ -239,6 +329,7 @@ def test_markdown_renderer_xss_protection(text):
assert "<script>" not in result
assert "<script>" in result
+
@given(content=st.text())
def test_markdown_renderer_headers(content):
if content and "\n" not in content:
diff --git a/tests/backend/test_telemetry_extended.py b/tests/backend/test_telemetry_extended.py
index bdc8e574..5550149c 100644
--- a/tests/backend/test_telemetry_extended.py
+++ b/tests/backend/test_telemetry_extended.py
@@ -1,4 +1,5 @@
import time
+
from meshchatx.src.backend.telemetry_utils import Telemeter
diff --git a/tests/backend/test_telemetry_integration.py b/tests/backend/test_telemetry_integration.py
index 0e825225..12e69118 100644
--- a/tests/backend/test_telemetry_integration.py
+++ b/tests/backend/test_telemetry_integration.py
@@ -1,6 +1,8 @@
-import pytest
import time
from unittest.mock import MagicMock
+
+import pytest
+
from meshchatx.meshchat import ReticulumMeshChat
from meshchatx.src.backend.telemetry_utils import Telemeter
@@ -49,7 +51,9 @@ async def test_process_incoming_telemetry_single(mock_app):
mock_lxmf_message.hash = b"msg_hash"
mock_app.process_incoming_telemetry(
- source_hash, packed_telemetry, mock_lxmf_message
+ source_hash,
+ packed_telemetry,
+ mock_lxmf_message,
)
# Verify database call
@@ -108,14 +112,15 @@ async def test_telemetry_request_parsing(mock_app):
# Bind on_lxmf_delivery
mock_app.on_lxmf_delivery = ReticulumMeshChat.on_lxmf_delivery.__get__(
- mock_app, ReticulumMeshChat
+ mock_app,
+ ReticulumMeshChat,
)
# Mocking dependencies
mock_app.is_destination_blocked.return_value = False
mock_app.current_context.config.telemetry_enabled.get.return_value = True
mock_app.database.contacts.get_contact_by_identity_hash.return_value = {
- "is_telemetry_trusted": True
+ "is_telemetry_trusted": True,
}
mock_app.database.messages.get_lxmf_message_by_hash.return_value = {} # To avoid JSON error
@@ -124,7 +129,7 @@ async def test_telemetry_request_parsing(mock_app):
# Verify handle_telemetry_request was called
mock_app.handle_telemetry_request.assert_called_with(
- "736f757263655f686173685f6279746573"
+ "736f757263655f686173685f6279746573",
)
@@ -135,4 +140,3 @@ async def test_tracking_toggle_endpoint(mock_app):
# We can't easily test the web endpoint here without more setup,
# but we can test the logic it calls if it was refactored into a method.
- pass
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────